package com.study.spark.scala.streaming

import java.sql.DriverManager

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * nc -lk 9999
  */
object NetworkWordCount {

  def main(args: Array[String]): Unit = {
    // Create a local StreamingContext with two working thread and batch interval of 10 second.
    // The master requires 2 cores to prevent a starvation scenario.
    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
    val ssc = new StreamingContext(conf, Seconds(10))
    // OR
    //val spark = SparkSession.builder().appName("NetworkWordCount").master("local[2]").getOrCreate()
    //val ssc = new StreamingContext(spark.sparkContext, Seconds(10))

    // Create a DStream that will connect to hostname:port, like localhost:9999
    // 创建SocketInputDStream
    val lines = ssc.socketTextStream("localhost", 9999)

    // lines=>words=>pairs=>wordCounts=>print
    // Split each line into words
    val words = lines.flatMap(_.split(" "))
    // Count each word in each batch
    val pairs = words.map(word => (word, 1))
    val wordCounts = pairs.reduceByKey(_ + _)
    // Print the first ten elements of each RDD generated in this DStream to the console
    wordCounts.print()


    //结果写入MySQL
    // 方案一：connection在driver端创建，在worker端执行，会报错
    //    wordCounts.foreachRDD { rdd =>
    //      val connection = createConnection()  // executed at the driver
    //      rdd.foreach { record =>
    //        connection.send(record) // executed at the worker
    //      }
    //    }

    // 方案二：存在每一条记录都会创建connection的问题
    //    result.foreachRDD { rdd =>
    //      rdd.foreach { record =>
    //        val connection = createConnection()
    //        connection.send(record)
    //        connection.close()
    //      }
    //    }

    //    // 方案三：在Partition中创建连接
    //    wordCounts.foreachRDD(rdd => {
    //      rdd.foreachPartition(partitionOfRecords => {
    //        // 在partition中创建connection
    //        val connection = createConnection();
    //        partitionOfRecords.foreach(record => {
    //          val sql = "insert into tb_word_count(word,word_count) values('" + record._1 + "'," + record._2 + ")"
    //          println(sql)
    //          // 存储数据
    //          connection.createStatement().execute(sql)
    //        })
    //        connection.close()
    //      })
    //    })


//    // DStream=>RDD=>DataFrame
//    wordCounts.foreachRDD(rdd => {
//      import spark.implicits._
//      val df = rdd.map(t => Record(t._1, t._2)).toDF()
//      df.createOrReplaceTempView("wc")
//      val rs = spark.sql("select * from wc")
//      rs.show()
//    })


    ssc.start() // Start the computation
    ssc.awaitTermination() // Wait for the computation to terminate
  }

  /** Case class for converting RDD to DataFrame */
  case class Record(word: String, count: Int)

  /**
    * 获取MySQL连接
    * 可以改为连接池
    *
    * @return
    */
  def createConnection() = {
    Class.forName("com.mysql.jdbc.Driver")
    DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "123456")
  }
}
